package com.yykj.hospital.server.message;

import com.google.common.reflect.TypeToken;
import com.yykj.common.util.GsonUtil;
import com.yykj.hospital.server.dto.account.IncomeDTO;
import com.yykj.hospital.server.enums.AccountLogType;
import com.yykj.hospital.server.service.account.AccountService;
import com.yykj.message.center.client.dto.TDMQMessage;
import com.yykj.message.center.client.dto.pay.PayMessageBodyDTO;
import com.yykj.message.center.client.enums.OrderType;
import com.yykj.user.center.client.dto.ConsultationOrderInfoDTO;
import com.yykj.user.center.client.feign.UserFeignClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class HospitalMessageService {

    //从配置文件中获取 serviceURL 接入地址、Token 密钥、Topic 全名和 Subscription 名称（均可从控制台复制）
    @Value("${tdmq.serviceUrl}")
    private String serviceUrl;
    @Value("${tdmq.token}")
    private String token;

    @Value("${tdmq.namespace}")
    private String namespace;

    private String prefix = "persistent://pulsar-pjnbmw52podb/";

    private PulsarClient client;

    private List<Consumer<byte[]>> consumerList = new ArrayList<>();

    private Consumer<byte[]> consumer;

    MessageListener<byte[]> messageListener;

    private UserFeignClient userFeignClient;

    @PostConstruct
    public void init() throws Exception {
        messageListener = new MessageListener<byte[]>() {
            @Override
            public void received(Consumer<byte[]> consumer, Message<byte[]> message) {
                try {
                    if (onMessage(message)) {
                        consumer.acknowledge(message);
                    } else {
                        consumer.negativeAcknowledge(message);
                    }
                } catch (PulsarClientException e) {
                    consumer.negativeAcknowledge(message);
                    e.printStackTrace();
                }
            }
        };

        client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(AuthenticationFactory.token(token))
                .build();
        System.out.println(">> pulsar client created.");
        //创建消费者

        consumer = client.newConsumer()
                .topic(prefix + namespace + "/pay-success")
                .subscriptionName("pay-success-client")
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .messageListener(messageListener)
                .subscribe();
        consumerList.add(consumer);

        consumer = client.newConsumer()
                .topic(prefix + namespace + "/task-finish")
                .subscriptionName("task-finish-client")
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(messageListener)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();
        consumerList.add(consumer);

    }

    @PreDestroy
    public void destroy() throws Exception {
        for (Consumer<byte[]> consumer : consumerList) {
            if (consumer != null) {
                consumer.close();
            }
            if (client != null) {
                client.close();
            }
        }

    }

    /**
     * 这里写消费业务逻辑
     *
     * @param message
     * @return return true: 消息ack  return false: 消息nack
     * @throws Exception 消息nack
     */
    private boolean onMessage(Message<byte[]> message) {
        //业务逻辑,延时类操作
        String msg = new String(message.getValue());
        log.info("收到消息：{}", msg);
        try {

            TDMQMessage tdmqMessage = GsonUtil.fromJson(new String(message.getValue()), TDMQMessage.class);
            ConsultationOrderInfoDTO consultationOrderInfoDTO;
            switch (tdmqMessage.getType()) {
                case PAY_SUCCESS:
                    TDMQMessage payMessage = GsonUtil.fromJson(new String(message.getValue()), new TypeToken<TDMQMessage<PayMessageBodyDTO>>(){}.getType());
                    PayMessageBodyDTO payMessageBodyDTO = (PayMessageBodyDTO)payMessage.getData();
                    if(OrderType.consultation.equals(payMessageBodyDTO.getOrderType())){
                        consultationOrderInfoDTO = userFeignClient.orderInfo(payMessageBodyDTO.getOrderId());
                        accountService.paySuccessAccount(IncomeDTO.builder()
                                .hospitalId(consultationOrderInfoDTO.getHospitalId())
                                .doctorId(consultationOrderInfoDTO.getDoctorId())
                                .patientId(consultationOrderInfoDTO.getPayAccountId())
                                .orderNum(consultationOrderInfoDTO.getOrderNo())
                                .logType(getAccountLogType(consultationOrderInfoDTO.getOrderType()))
                                .price(consultationOrderInfoDTO.getPayAmount())
                                .detail("用户支付")
                                .build());
                    }else{
                        log.info("不是需要的消息 跳过：{}", msg);
                        return true;
                    }
                    break;
                case TASK_FINISHED:
                    TDMQMessage finishMessage = GsonUtil.fromJson(new String(message.getValue()), new TypeToken<TDMQMessage<PayMessageBodyDTO>>(){}.getType());
                    PayMessageBodyDTO messageBodyDTO = (PayMessageBodyDTO)finishMessage.getData();
                    if(OrderType.consultation.equals(messageBodyDTO.getOrderType())){
                        consultationOrderInfoDTO = userFeignClient.orderInfo(messageBodyDTO.getOrderId());
                        accountService.doctorFinishTaskIncome(IncomeDTO.builder()
                                .hospitalId(consultationOrderInfoDTO.getHospitalId())
                                .doctorId(consultationOrderInfoDTO.getDoctorId())
                                .patientId(consultationOrderInfoDTO.getPayAccountId())
                                .orderNum(consultationOrderInfoDTO.getOrderNo())
                                .logType(getAccountLogType(consultationOrderInfoDTO.getOrderType()))
                                .price(consultationOrderInfoDTO.getPayAmount())
                                .detail("任务分红")
                                .build());
                    }else{
                        log.info("不是需要的消息 跳过：{}", msg);
                        return true;
                    }

                    break;
                case DOCTOR_INFO_CHANGE:
                    break;
                default:
                    break;
            }
            log.info("消息处理 succeed：{}", msg);
            return true;
        } catch (Exception exception) {
            log.error("消息处理 failed：{}", msg);
        }
        return false;
    }

    private AccountService accountService;

    @Autowired
    public void setUserFeignClient(UserFeignClient userFeignClient) {
        this.userFeignClient = userFeignClient;
    }

    @Autowired
    public void setAccountService(AccountService accountService) {
        this.accountService = accountService;
    }

    private AccountLogType getAccountLogType(Integer orderType) {
        switch (orderType) {
            case 11:
                return AccountLogType.pic;
            case 12:
                return AccountLogType.voice;
            case 13:
                return AccountLogType.video;
            default:
                return null;
        }
    }
}
